package org.anno.chunkdb.backend;

import org.anno.chunkdb.Chunkdb;
import org.anno.chunkdb.backend.query.NumberResultColumn;

import java.util.Comparator;
import java.util.function.DoubleConsumer;
import java.util.function.ToIntFunction;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;

/**
 * Stores double values in an array.
 */
public class DoubleColumn extends IndexedColumn implements Table.Column<FlyRow>,
                                                           NumberResultColumn<FlyRow> {

  private final double[] data;

  public DoubleColumn(double[] data, int[] index) {
    super(index);
    this.data = data;
  }

  public Comparator<FlyRow> getComparator() {
    return Comparator.comparingDouble(this::getDouble);
  }

  public ToIntFunction<FlyRow> getDiffFunction(Chunkdb.Filter filter) {
    double rhs = filter.getDoubleRhs();
    return r -> Double.compare(getDouble(r), rhs);
  }

  public double getDouble(FlyRow row) {
    return data[row.i];
  }

  /**
   * Returns smallest i s.t. i == N || v[index[i]] >= v. Example: rank(x, {x,x}) == 0,
   * rank(x+1,{x,x}) == 2.
   */
  public int getGERank(double v) {
    int start = 0;
    int end = index.length;
    while (start < end) {
      int mid = (start + end) / 2;
      if (data[index[mid]] < v) {
        start = mid + 1;
      } else {
        end = mid;
      }
    }
    return start;
  }

  /**
   * Returns smallest i s.t. i == N || v[index[i]] > v. Example: rank(x, {x,x}) == 2,
   * rank(x-1,{x,x}) == 0.
   */
  public int getGTRank(double v) {
    int start = 0;
    int end = index.length;
    while (start < end) {
      int mid = (start + end) / 2;
      if (data[index[mid]] <= v) {
        start = mid + 1;
      } else {
        end = mid;
      }
    }
    return start;
  }

  /**
   * Returns largest i s.t. i == -1 || v[index[i]] <= v. Example: rank(x, {x,x}) == 1,
   * rank(x-1,{x,x}) == -1.
   */
  public int getLERank(double v) {
    int start = -1;
    int end = index.length - 1;
    while (start < end) {
      int mid = (start + end + 1) / 2;
      if (data[index[mid]] <= v) {
        start = mid;
      } else {
        end = mid - 1;
      }
    }
    return start;
  }

  /**
   * Returns largest i s.t. i == -1 || v[index[i]] < v. Example: rank(x,{x,x}) == -1, rank(x+1,
   * {x,x}) == 1.
   */
  public int getLTRank(double v) {
    int start = -1;
    int end = index.length - 1;
    while (start < end) {
      int mid = (start + end + 1) / 2;
      if (data[index[mid]] < v) {
        start = mid;
      } else {
        end = mid - 1;
      }
    }
    return start;
  }

  public static Builder builder() {
    return new Builder();
  }

  public static class Builder implements DoubleConsumer {

    private DoubleStream.Builder builder = DoubleStream.builder();

    public DoubleColumn build() {
      double[] data = builder.build().toArray();
      int[] index = IntStream.range(0, data.length)
          .mapToObj(FlyRow::new)
          .sorted(Comparator.comparing(r -> data[r.i]))
          .mapToInt(r -> r.i)
          .toArray();
      return new DoubleColumn(data, index);
    }

    public void accept(double value) {
      builder.accept(value);
    }
  }
}
